package oss

import (
	"bytes"
	"context"
	"fmt"
	"gitee.com/ymofen/gobase"
	"gitee.com/ymofen/gobase/factory"
	"gitee.com/ymofen/gobase/subpub"
	"runtime"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

// 存储数据到OSS中
type OssProducer struct {
	startedflag int32
	taskId      string
	lk          sync.RWMutex
	caption     string
	buketName   string
	confMap     gobase.StrMap
	ex          *subpub.Subchannel

	op OSSOperator

	recvtimout time.Duration
	lastrecvT  time.Time
	statusText string
}

var (
	OssProducerN      atomic.Int32
	OssProducerAliveN atomic.Int32
)

func (this *OssProducer) Close() error {
	OssProducerN.Add(-1)
	this.lk.Lock()
	defer this.lk.Unlock()
	this.confMap = nil
	this.ex = nil
	this.taskId = ""
	this.startedflag = 0
	if this.op != nil {
		gobase.TryClose(this.op)
		this.op = nil
	}
	return nil
}

func (this *OssProducer) GetConnString() string {
	return ""
}

func (this *OssProducer) Pub(topic string, max int, args ...interface{}) int {
	if len(args) == 0 {
		return -1
	}

	var argsBuf []byte

	if buf, ok := args[0].([]byte); ok {
		argsBuf = buf
	} else if str, ok := args[0].(string); ok {
		argsBuf = []byte(str)
	} else {
		logger.Warnf("%s\toss publish args convert fail\tdata=%v", this.caption, args[0])
		return -1
	}

	bufR := bytes.NewReader(argsBuf)

	err := this.op.PutObject(context.Background(), this.buketName, topic, bufR)
	if err != nil {
		logger.Warnf("%s\toss publish fail\terror=%s", this.caption, err.Error())
		return -1
	}
	return 1
}

func (this *OssProducer) HealthyCheck() (healthy gobase.Ternary, msg string) {
	if atomic.LoadInt32(&this.startedflag) == 0 {
		return gobase.TernaryFalse, "shutdown"
	}
	this.lk.RLock()
	defer this.lk.RUnlock()
	if this.op == nil {
		return gobase.TernaryFalse, "oss operator is nil"
	}

	return gobase.TernaryFalse, "shutdown"
}

func (this *OssProducer) GetStatus(args ...interface{}) string {
	this.lk.RLock()
	defer this.lk.RUnlock()
	if this.startedflag == 0 {
		return fmt.Sprintf("shutdown")
	}
	return fmt.Sprintf("%s&ex=%s&status=%s", this.confMap.URLEncode0(), this.ex.Status(), this.statusText)
}

func (this *OssProducer) CheckStart() error {
	this.lk.Lock()
	defer this.lk.Unlock()
	if this.startedflag == 1 {
		return nil
	}
	defer func() {
		this.startedflag = 1
	}()

	sn := OssProducerN.Add(1)

	this.caption = this.confMap["name"]
	if len(this.caption) == 0 {
		this.caption = fmt.Sprintf("%s-p%d", this.confMap["stream"], sn)
	}

	this.buketName = this.confMap["buketName"]

	implId := this.confMap["impl"]
	if len(implId) == 0 {
		return fmt.Errorf("invalid param, impl is empty")
	}
	implId = fmt.Sprintf("oss.%s", implId)

	ossConf := gobase.NewStrMap()
	ossConf.CopyFromPrefix(this.confMap, "oss.", true)

	obj, err := gobase.CreateFactoryInstance(implId, ossConf)
	if err != nil {
		return err
	}
	this.op = obj.(OSSOperator)

	this.taskId = fmt.Sprintf("OssProducer-%d", sn)

	return nil
}

var (
	DefaultOssProducers = gobase.NewConnectObjectRefStorage(newOssProducer)
)

func newOssProducer(connstr string) interface{} {
	node := &OssProducer{
		confMap: gobase.NewStrMap(),
		ex:      subpub.NewSubchannel(),
	}
	OssProducerAliveN.Add(1)
	runtime.SetFinalizer(node, func(obj interface{}) {
		OssProducerAliveN.Add(-1)
	})
	node.confMap.URLFormDecode(connstr)

	return node
}

// conntype=ossproducer&oss.impl=s3client
func checkGetOssProducer(args ...interface{}) (rval interface{}, err error) {
	if len(args) == 0 {
		err = fmt.Errorf("无效的参数")
		return
	}
	if conf, ok := args[0].(gobase.StrMap); ok {
		s := conf.SortRangeUrlEncode(func(k, v string) bool {
			return k == "buketName" || k == "name" || k == "connrand" || k == "impl" || strings.HasPrefix(k, "oss.")
		})
		rval = DefaultOssProducers.CheckGetRef(s)
		return
	} else {
		err = fmt.Errorf("无效的参数")
		return
	}
}

func init() {
	factory.RegisterCreateFunc("sp.ossproducer", checkGetOssProducer)
}
